package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description:
 * @Author: JK
 * @Date: create in 2021/7/29 16:54
 */
@Slf4j
@Component
public class PromoOrderProducer {

    TransactionMQProducer producer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object args) {
                // 扣减秒杀商品库存  psId  productId
                Map<String, Long> paramMap = (Map<String, Long>) args;
                Long psId = paramMap.get("psId");
                Long productId = paramMap.get("productId");
                System.out.println("psId = " + psId);
                System.out.println("productId = " + productId);

                // 在Redis中保存本地事务的执行结果的key，通过transactionId，作为本地事务的唯一表示
                String transactionKey = "promo_order_local_transaction_" + message.getTransactionId();

                // 在Redis中保存秒杀商品是否售罄的标志，从而在商品售罄狗无需执行后面的步骤，提高效率
                String StockNotEnough = "stock_not_enough_" + psId + "_" + productId;

                // 使用基于redis的分布式锁
                String lockKey = "promo_item_stock_lock_" + psId + "_" + productId;
                RLock distributedLock = redissonClient.getLock(lockKey);

                // 如果加锁失败，失败的线程会阻塞等待，相当于对扣减库存做了串行化操作
                distributedLock.lock();
                try {
                    Integer affectedRows = promoItemMapper.decreaseStock(productId, psId);
                    if (affectedRows < 1) {
                        log.info("秒杀商品库存扣减失败！");
                        // 在Redis中保存商品售罄的标志
                        cacheManager.setCache(StockNotEnough, "stockNotEnough", 1);
                        // 在Redis中保存本地事务执行结果
                        cacheManager.setCache(transactionKey, "fail", 1);
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                } finally {
                    // 使用try_finally确保锁一定会被释放
                    distributedLock.unlock();
                }

                log.info("秒杀商品库存扣减成功！");
                // 在Redis中保存本地事务执行结果
                cacheManager.setCache(transactionKey, "success", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt message) {
                // 在Redis中保存本地事务的执行结果的key，通过transactionId，作为本地事务的唯一表示
                String transactionKey = "promo_order_local_transaction_" + message.getTransactionId();

                String transactionResult = cacheManager.checkCache(transactionKey);
                if (transactionResult != null && "success".equals(transactionResult)) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

                if (transactionResult != null && "fail".equals(transactionResult)) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                return LocalTransactionState.UNKNOW;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public boolean sendCreatePromoOrderMessage(CreateSeckillOrderRequest request, Long psId, Long productId) {
        // 1.准备发送事务消息
        String s = JSON.toJSONString(request);
        Message message = new Message("create_promotion_order", s.getBytes(StandardCharsets.UTF_8));

        // 准备本地事务执行所需的参数
        Map<String, Long> paramMap = new HashMap<>();
        paramMap.put("psId", psId);
        paramMap.put("productId", productId);

        // 2.发送事务消息
        TransactionSendResult sendResult = null;
        try {
            sendResult = producer.sendMessageInTransaction(message, paramMap);
            if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
                // 消息发送成功
                log.info("sendCreatePromoOrderMessage 消息发送成功，且本地事务执行成功！");
                return true;
            } else if (sendResult != null && LocalTransactionState.ROLLBACK_MESSAGE.equals(sendResult.getLocalTransactionState())) {
                // 本地事务执行失败
                log.info("sendCreatePromoOrderMessage 消息发送成功，但本地事务执行失败！");
                return false;
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        return false;
    }
}
